package com.xxk.rocket.star.core.consume;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;

public class ConsumeHandlerImpl {

  private static final Logger log = LoggerFactory.getLogger(ConsumeHandlerImpl.class);

  private ExecutorService executorService;
  private Semaphore semaphore;

  public ConsumeHandlerImpl() {
    this(4);
  }

  public ConsumeHandlerImpl(int threadCount) {
    semaphore = new Semaphore(threadCount, true);
    this.executorService = Executors.newFixedThreadPool(threadCount);
  }

  public Future<ConsumeResult> addConsumeRunnable(ConsumeHandlerRunnable consumeHandlerRunnable)
          throws InterruptedException {
    semaphore.acquire();
    consumeHandlerRunnable.setSemaphore(semaphore);
    Future<ConsumeResult> future = executorService.submit(consumeHandlerRunnable);
    return future;
  }

  public void shutdown() {
    while (semaphore.hasQueuedThreads()) {
      try {
        semaphore.acquire();
        semaphore.release();
        Thread.sleep(100);
      } catch (InterruptedException e) {
        //will not happen
        e.printStackTrace();
      }
    }
    executorService.shutdown();
  }

  public boolean isShutdown() {
    return executorService.isShutdown();
  }
}
